package my.thesis.bolts.computations;

import java.util.Map;

import my.thesis.analytics.metrics.LastTimePersistMetric;
import my.thesis.analytics.metrics.ThroughputMetric;
import my.thesis.analytics.metrics.whereAmIMetric;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class TimeIntervalExtractor extends BaseRichBolt 
{
	private static final long serialVersionUID = 1L;
	private OutputCollector collector;
	private int tiSec;
	
	transient ThroughputMetric _throughputMetric;
	private static Boolean metricsOn=false;

	public TimeIntervalExtractor(int sec)
	{
		tiSec=sec;
	}
	
	public void prepare(Map stormConf, TopologyContext context,OutputCollector collector) 
	{
		this.collector=collector;		
		
		if(metricsOn)
		{
			_throughputMetric=new ThroughputMetric();
			context.registerMetric("timeIntervalCount", _throughputMetric, 20);
			_throughputMetric.setStartTime();
		}
	}

	public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) 
	{		 
			outputFieldsDeclarer.declare(new Fields("id","millis","lang","hash","timeInterval"));
	}
	
	private Long selectTimeInterval(Long time) 
	{
		    return time / (tiSec * 1000);
	}

	public void execute(Tuple tuple) 
	{
		if(metricsOn)
		{
			_throughputMetric.incrTuples();
		}
		
		Long id = tuple.getLongByField("id");
		String hash = tuple.getStringByField("hash");
		String lang = tuple.getStringByField("lang");
		long millis = tuple.getLongByField("millis");
		long timeIntervalMillis = selectTimeInterval(millis);

		//System.out.println("EMIT TI:" + timeIntervalMillis + "  HASH:" + hash);
		collector.emit(new Values(id,millis,lang,hash,timeIntervalMillis));

	}
}
